package net.sina.realtime.traffic.job;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import net.sina.realtime.traffic.utils.DateTimeUtil;
import net.sina.realtime.traffic.utils.JdbcUtil;
import net.sina.realtime.traffic.utils.KafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

public class JobTrafficWfTotalJob {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //  数据源
        DataStream<String> trafficStream = KafkaUtil.consumerKafka(env, "traffic-log");

        //违法事件存储
        illegalEvents(trafficStream);

        //违法类型统计存储
        illegalType(trafficStream);

        env.execute("JobTrafficWfTotalJob");
    }

    private static void illegalType(DataStream<String> trafficStream) {
        SingleOutputStreamOperator<String> filterStream = trafficStream.filter(
                line -> !JSON.parseObject(line).getString("event_type")
                        .equals("normal")
        );

        SingleOutputStreamOperator<String> watermarksStream = filterStream.assignTimestampsAndWatermarks(
                WatermarkStrategy.<String>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                        .withTimestampAssigner(new SerializableTimestampAssigner<String>() {
                            @Override
                            public long extractTimestamp(String element, long recordTimestamp) {
                                return JSON.parseObject(element).getLong("timestamp");
                            }
                        })
        );

        KeyedStream<String, String> keyByStream = watermarksStream.keyBy(
                line -> JSON.parseObject(line).getString("event_type")
        );

        WindowedStream<String, String, TimeWindow> windowStream = keyByStream.window(
                TumblingEventTimeWindows.of(Time.minutes(1))
        );

        SingleOutputStreamOperator<String> illegalTypeStream = windowStream.process(
                new ProcessWindowFunction<String, String, String, TimeWindow>() {
                    @Override
                    public void process(String s, Context context, Iterable<String> elements, Collector<String> out) throws Exception {
                        long count = 0L;
                        for (String element : elements) {
                            count++;
                        }
                        String window_start = DateTimeUtil.convertLongToString(context.window().getStart(),DateTimeUtil.DATE_TIME_FORMAT);
                        String window_end = DateTimeUtil.convertLongToString(context.window().getEnd(),DateTimeUtil.DATE_TIME_FORMAT);
                        long ts = System.currentTimeMillis();
                        String output =window_start + "," + window_end + "," + s + "," + count + "," + ts;
                        out.collect(output);
                    }
                });

        JdbcUtil.sinkToClickhouseUpsert(
                illegalTypeStream,
                "INSERT INTO TABLE traffic_log_report.dwd_traffic_illegal_type\n" +
                        "(\n" +
                        "    window_start, window_end, eventType, count, ts\n" +
                        ") VALUES ( ?, ?, ?, ?, ? )"
        );
    }

    private static void illegalEvents(DataStream<String> trafficStream) {
        SingleOutputStreamOperator<String> filterStream = trafficStream.filter(
                line -> !JSON.parseObject(line).getString("event_type")
                        .equals("normal")
        );

        SingleOutputStreamOperator<String> illegalEventsStream = filterStream.map(new MapFunction<String, String>() {
            @Override
            public String map(String value) throws Exception {
                JSONObject jsonObject = JSON.parseObject(value);
                return jsonObject.getString("license_plate") + "," +
                        jsonObject.getString("road_id") + "," +
                        jsonObject.getInteger("lane") + "," +
                        jsonObject.getString("event_type")+","+
                        jsonObject.getLong("timestamp");
            }
        });

        JdbcUtil.sinkToClickhouseUpsert(
                illegalEventsStream,
                "INSERT INTO TABLE traffic_log_report.dwd_traffic_illegal_events\n" +
                        "(\n" +
                        "    licensePlate, roadId, lane, eventType, ts\n" +
                        ") VALUES ( ?, ?, ? , ?, ?)"
        );
    }

}
